﻿/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include "hadoop/SerialUtils.hh"
#include "hadoop/StringUtils.hh"

#include <errno.h>
#include <QtCore/QFile>
#include <glog/logging.h>
#ifdef OS_TYPE_Windows
#else
#include <rpc/types.h>
#include <rpc/xdr.h>
#endif
#include <string>
#include <string.h>



namespace HadoopUtils {

  Error::Error(const std::string& msg): std::runtime_error(msg),error(msg) {
  }

  Error::Error(const std::string& msg, 
               const std::string& file, int line, 
               const std::string& function): std::runtime_error(msg){
    error = msg + " at " + file + ":" + toString(line) + 
            " in " + function;
  }

  const std::string& Error::getMessage() const {
    return error;
  }

  FileInStream::FileInStream()
  {
    mFile = NULL;
    isOwned = false;
  }

  bool FileInStream::open(const std::string& name)
  {
	 
    mFile = new QFile(name.c_str());
	if(!mFile->open(QIODevice::ReadOnly)){
		delete mFile;
		mFile=NULL;
		return false;
	}
    isOwned = true;
    return (mFile != NULL);
  }

  bool FileInStream::open(QIODevice* file)
  {
    mFile = file;
    isOwned = false;
    return (mFile != NULL);
  }

  void FileInStream::read(void *buf, size_t len)
  {
	   const int Timeout = 50 * 1000;
	  while (true) {
		  int avail=this->mFile->bytesAvailable();
		  if(avail >= (qint64 ) len) break;
		  if (!this->mFile->waitForReadyRead(Timeout)) {
			  LOG(ERROR)<<"read fail,end of file,需要读"<<len<<"字节,但是一直读不到数据，放弃";
			  throw HadoopUtils::Error("read fail");
			  return;
		  }
	  }

	 QByteArray qb=this->mFile->read(len);
	qint64 result=qb.length();	
    if (result == 0 || result!=len) {
      if (mFile->atEnd()) {
		LOG(ERROR)<<"read fail,end of file,需要读"<<len<<"字节,但是读了"<<result<<"字节后就遇到文件终止";
        throw HadoopUtils::Error("read fail");
      } else {
        HADOOP_ASSERT(false, std::string("read error on file: ") + strerror(errno));
      }
    }	
	memcpy(buf,qb.data(),result);
  }

  bool FileInStream::skip(size_t nbytes)
  {
	  qint64 pos=this->mFile->pos();
	  return this->mFile->seek(pos+nbytes);
  }

  bool FileInStream::close()
  {
    int ret = 0;
    if (mFile != NULL && isOwned) {
		this->mFile->close();
		delete mFile;
    }
    mFile = NULL;
    return (ret==0);
  }

  FileInStream::~FileInStream()
  {
    if (mFile != NULL) {
      close();
    }
  }

  FileOutStream::FileOutStream()
  {
    mFile = NULL;
    isOwned = false;
  }

  bool FileOutStream::open(const std::string& name, bool overwrite)
  {
    if (!overwrite) {
      mFile = new QFile(QString::fromStdString(name));
	  if(!mFile->open(QIODevice::ReadOnly)){
		  delete mFile;
		  return false;
	  }
    }
    mFile = new QFile(QString::fromStdString(name));
	bool ret =mFile->open(QIODevice::WriteOnly|QIODevice::Truncate);
    isOwned = true;
    return ret;
  }

  bool FileOutStream::open(QIODevice* file)
  {
	  mFile = file;
	  isOwned = false;
	  return (mFile != NULL);
  }

  void FileOutStream::write(const void* buf, size_t len)
  {
	  size_t remain=len;
	  const char* p=(const char*)buf;
	  while(remain>0){
		const qint64 ret=this->mFile->write(p,remain);
		if(ret>0){
			remain-=ret;
			p+=ret;
		} else if(ret<0){
			QString errstr=this->mFile->errorString();
			LOG(ERROR)<<"write error to file: ret="<<ret<<",error message="<<errstr.toStdString();
			throw  std::runtime_error("error while writing data");
		} else {
			this->mFile->waitForBytesWritten(5000);	
		}	 
	  }
  }

  bool FileOutStream::advance(size_t nbytes)
  {
	  qint64 pos=this->mFile->pos();
	  return this->mFile->seek(pos+nbytes);
  }

  bool FileOutStream::close()
  {
    int ret = 0;
    if (mFile != NULL && isOwned) {
		this->mFile->close();
		delete mFile;
    }
    mFile = NULL;
    return (ret == 0);
  }

  void FileOutStream::flush()
  {
    //this->mFile->
	//TODO:
  }

  FileOutStream::~FileOutStream()
  {
    if (mFile != NULL) {
      close();
    }
  }

  StringInStream::StringInStream(const std::string& str): buffer(str) {
    itr = buffer.begin();
  }

  void StringInStream::read(void *buf, size_t buflen) {
    size_t bytes = 0;
    char* output = (char*) buf;
    std::string::const_iterator end = buffer.end();
    while (bytes < buflen) {
      output[bytes++] = *itr;
      ++itr;
      if (itr == end) {
        break;
      }
    }
    HADOOP_ASSERT(bytes == buflen, "unexpected end of string reached");
  }

  void StringOutStream::write(const void *buf, size_t len) {
	  this->data.append((const char*)buf,len);
  }

  std::string StringOutStream::str(){
	  return this->data;
  }

  void serializeInt(int32_t t, OutStream& stream) {
    serializeLong(t,stream);
  }

  void serializeLong(int64_t t, OutStream& stream)
  {
    if (t >= -112 && t <= 127) {
      int8_t b = static_cast<int8_t>(t);
      stream.write(&b, 1);
      return;
    }
        
    int8_t len = -112;
    if (t < 0) {
      t ^= -1ll; // reset the sign bit
      len = -120;
    }
        
    uint64_t tmp = t;
    while (tmp != 0) {
      tmp = tmp >> 8;
      len--;
    }
  
    stream.write(&len, 1);      
    len = (len < -120) ? -(len + 120) : -(len + 112);
        
    for (uint32_t idx = len; idx != 0; idx--) {
      uint32_t shiftbits = (idx - 1) * 8;
      uint64_t mask = 0xFFll << shiftbits;
      uint8_t b = static_cast<int8_t>((t & mask) >> shiftbits);
      stream.write(&b, 1);
    }
  }

  int32_t deserializeInt(InStream& stream) {
    return static_cast<int32_t>(deserializeLong(stream));
  }

  int64_t deserializeLong(InStream& stream)
  {
    int8_t b;
    stream.read(&b, 1);
    if (b >= -112) {
      return b;
    }
    bool negative;
    int len;
    if (b < -120) {
      negative = true;
      len = -120 - b;
    } else {
      negative = false;
      len = -112 - b;
    }
    uint8_t* barr=new uint8_t[len];
	try{
    stream.read(barr, len);
	}catch(...){
		delete[] barr;
		throw;
	}
    int64_t t = 0;
    for (int idx = 0; idx < len; idx++) {
      t = t << 8;
      t |= (barr[idx] & 0xFF);
    }
	delete[] barr;
    if (negative) {
      t ^= -1ll;
    }
    return t;
  }
#include <QtCore/QtEndian>
  void serializeFloat(float t, OutStream& stream)
  {
	  uint32_t tmp = *(uint32_t*)&t;
	  uint32_t ret;
	  qToBigEndian(tmp,(uchar*)&ret);
	  stream.write(&ret,sizeof(ret));  
  }

  float deserializeFloat(InStream& stream){	  
	  char buf[sizeof(float)];
	  stream.read(buf, sizeof(float));
	  uint32_t tmp=*(uint32_t*)buf;
	  tmp=qFromBigEndian(tmp);
	  return *(float*)&tmp;	
  }
  

  void serializeString(const std::string& t, OutStream& stream)
  {
    serializeInt(t.length(), stream);
    if (t.length() > 0) {
      stream.write(t.data(), t.length());
    }
  }

  void deserializeString(std::string& t, InStream& stream)
  {
    int32_t len = deserializeInt(stream);
    if (len > 0) {
      // resize the string to the right length
      t.resize(len);
      // read into the string in 64k chunks
      const int bufSize = 65536;
      int offset = 0;
      char buf[bufSize];
      while (len > 0) {
        int chunkLength = len > bufSize ? bufSize : len;
        stream.read(buf, chunkLength);
        t.replace(offset, chunkLength, buf, chunkLength);
        offset += chunkLength;
        len -= chunkLength;
      }
    } else {
      t.clear();
    }
  }

}
